package cn.gwm.flink.streaming.dws.reservecharge.test;

import cn.gwm.utils.DateFormatDefine;
import cn.gwm.utils.DateUtils;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.io.FileUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.*;
import java.util.ArrayList;

/**
 * @author Liu.Nanfang
 * @date 2022/7/15 13:46
 * @description 模拟生产数据
 */
public class CustomFileSource implements SourceFunction<String> {

    private Integer count = 1;

    private boolean isRunning = true;

    private ArrayList<String> fileList = new ArrayList<>();

    {
        try {
            setList();
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
//        saveDataAnalyze();
        CustomFileSource customFileSource = new CustomFileSource();
//        customFileSource.run();
    }

    /**
     * 导出指定数据
     */
    public static void saveDataAnalyze() {
        String textPath = "E:/Cache/reserve_new.json";
        String textPathOut = "E:/Cache/charge_test_file/reserve_new_out.json";
        File file = new File(textPath);
        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String tempString = null;
            int line = 1;
            while ((tempString = reader.readLine()) != null) {
                System.out.println("line " + line + ": " + tempString);
                JSONObject json = JSONObject.parseObject(tempString);
                String str = String.format("%s  %s  %s  %s  %s  %s  %s  %s  %s",
                        json.getString("vin"),
                        json.getString("adCode"),
//                        json.getBigDecimal("longitude"),
//                        json.getBigDecimal("longitude"),
                        json.getString("chargeStatus"),
                        json.getString("chargeType"),
                        DateUtils.getCurrentDateTime(DateFormatDefine.DATE_TIME_FORMAT, json.getDate("tid")),
                        json.getString("startChargeTime") != null ? DateUtils.getCurrentDateTime(DateFormatDefine.DATE_TIME_FORMAT, json.getDate("startChargeTime")) : null,
                        json.getString("endChargeTime") != null ? DateUtils.getCurrentDateTime(DateFormatDefine.DATE_TIME_FORMAT, json.getDate("endChargeTime")) : null,
                        json.getString("chrgConnectTid") != null ? DateUtils.getCurrentDateTime(DateFormatDefine.DATE_TIME_FORMAT, json.getDate("chrgConnectTid")) : null,
                        json.getString("endChrgConnectTid") != null ? DateUtils.getCurrentDateTime(DateFormatDefine.DATE_TIME_FORMAT, json.getDate("endChrgConnectTid")) : null
                );
                FileUtils.writeStringToFile(new File(textPathOut), str + "\n", true);
                line++;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    private void setList() throws FileNotFoundException {
        String textPath = "E:\\Cache\\reserve_charge\\testdata";
        File file = new File(textPath);
        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String tempString = null;
            int line = 1;
            // 一次读入一行，直到读入null为文件结束
            while ((tempString = reader.readLine()) != null) {
                System.out.println("line " + line + ": " + tempString);
                fileList.add(tempString);
                line++;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        while (isRunning) {
            sourceContext.collect(fileList.get(count));
            count++;
            if (count >= fileList.size()) {
//                count = 0;
//                continue;
                return;
            }
//            System.out.println("index: " + count);
            // 每秒产生一条数据
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
